package a_o2odata_deal.utils.zhibo_zf.zhengli

import org.apache.spark.sql.SQLContext
import a_aa_amainpackage.a_o2odata_deal.config.config._

/**
  * @ Auther: o2o-rd-0008
  * @ Date:   2020/2/11 10:53
  * @ Param:  ${PARAM}
  * @ Description: 
  */
object handle_good {
  def handle_good(sqlContext:SQLContext):Unit={
    val tb_data = sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/taobao/${years}/${months}/zhibo_finally")
      .selectExpr("good_id","sellCount","salesAmount","province","city","district","regional_ID",
        "firstCategoryId","secondCategoryId","thirdCategoryId","fourthCategoryId","platformId","timeStamp",
        "is_showLive","is_selfLive","viewcount")
    val tm_data = sqlContext.read.orc(s"s3a://o2o-dataproces-group/panzonghao/tmall/${years}/${months}/zhibo_finally")
      .selectExpr("good_id","sellCount","salesAmount","province","city","district","regional_ID",
        "firstCategoryId","secondCategoryId","thirdCategoryId","fourthCategoryId","platformId","timeStamp",
        "is_showLive","is_selfLive","viewcount")

    val source_data = tb_data.union(tm_data)
    source_data.createOrReplaceTempView("source_data")

    //计算每个区县每个三级行业下的总销量，销售额，商品条数
    sqlContext.sql(
      s"""
         |select
         |*,
         |sum(sellCount) over(partition by major_key) as sellCounts,
         |sum(salesAmount) over(partition by major_key) as salesAmounts,
         |count(good_id) over(partition by major_key) as goodCount
         |from
         |(select
         |*,
         |concat(concat(concat(province,city),district),thirdCategoryId) as major_key
         |from source_data)
         """.stripMargin).drop("sellCount","salesAmount").dropDuplicates("major_key")
      .withColumnRenamed("sellCounts","sellCount")
      .withColumnRenamed("salesAmounts","salesAmount")
      .createOrReplaceTempView("all_data")

    //将直播数据摘出来
    source_data.where("cast(is_showLive as boolean)=true").createOrReplaceTempView("zhibo_data")
    //计算每个区县每个三级行业下的总观看人数，直播商品数，直播总销量，直播总销售额
    sqlContext.sql(
      s"""
         |select
         |*,
         |sum(sellCount) over(partition by major_key) as live_sellCount,
         |sum(salesAmount) over(partition by major_key) as live_salesAmount,
         |sum(viewcount) over(partition by major_key) as viewCount,
         |count(good_id) over(partition by major_key) as live_goodCount
         |from
         |(select
         |*,
         |concat(concat(concat(province,city),district),thirdCategoryId) as major_key
         |from zhibo_data)
         """.stripMargin).dropDuplicates("major_key").createOrReplaceTempView("live_data")

    sqlContext.sql(
      s"""
         |select
         |t1.major_key,
         |t1.province,
         |t1.city,
         |t1.district,
         |t1.regional_ID,
         |t1.firstCategoryId,
         |t1.secondCategoryId,
         |t1.thirdCategoryId,
         |t1.fourthCategoryId,
         |t1.platformId,
         |t1.timeStamp,
         |t1.sellCount,
         |t1.salesAmount,
         |t1.goodCount,
         |case when t2.major_key is null then '0' else t2.live_sellCount end as live_sellCount,
         |case when t2.major_key is null then '0' else t2.live_salesAmount end as live_salesAmount,
         |case when t2.major_key is null then '0' else t2.viewCount end as viewCount,
         |case when t2.major_key is null then '0' else t2.live_goodCount end as live_goodCount
         |from all_data t1
         |left join
         |live_data t2
         |on t1.major_key=t2.major_key
         """.stripMargin).createOrReplaceTempView("tmp")

    sqlContext.sql(
      s"""
         |select
         |*,
         |cast(sellCount as bigint) as sellCounts,
         |cast(salesAmount as double) as salesAmounts,
         |cast(goodCount as bigint) as goodCounts,
         |cast(live_sellCount as string) as live_sellCounts,
         |cast(live_salesAmount as string) as live_salesAmounts,
         |cast(viewCount as string) as viewCounts,
         |cast(live_goodCount as string) as live_goodCounts
         |from tmp
         """.stripMargin).drop("sellCount","salesAmount","goodCount","live_sellCount","live_salesAmount","viewCount","live_goodCount")
      .withColumnRenamed("sellCounts","sellCount")
      .withColumnRenamed("salesAmounts","salesAmount")
      .withColumnRenamed("goodCounts","goodCount")
      .withColumnRenamed("live_sellCounts","live_sellCount")
      .withColumnRenamed("live_salesAmounts","live_salesAmount")
      .withColumnRenamed("viewCounts","viewCount")
      .withColumnRenamed("live_goodCounts","live_goodCount")

      .repartition(4).write.orc(s"s3a://o2o-dataproces-group/panzonghao/zhibo_zf/${years}/${months}/good")


  }
}
